package com.seedog.mq.core;

import com.seedog.mq.model.Message;
import com.seedog.mq.exception.TopicNotFoundException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 主题实现类，提供消息的存储和分发功能
 */
public class TopicImpl implements Topic {
    private static final Logger logger = LoggerFactory.getLogger(TopicImpl.class);
    
    private final String name;
    private final LinkedBlockingQueue<Message> messageQueue;
    private final Set<String> consumers;
    private final Map<String, Map<String, Message>> unacknowledgedMessages;
    private final AtomicInteger messageCount;
    private boolean closed = false;
    
    public TopicImpl(String name) {
        this.name = name;
        this.messageQueue = new LinkedBlockingQueue<>();
        this.consumers = ConcurrentHashMap.newKeySet();
        this.unacknowledgedMessages = new ConcurrentHashMap<>();
        this.messageCount = new AtomicInteger(0);
        logger.info("Topic '{}' created", name);
    }
    
    @Override
    public String getName() {
        return name;
    }
    
    @Override
    public synchronized String sendMessage(Message message) {
        checkClosed();
        
        if (message == null) {
            throw new IllegalArgumentException("Message cannot be null");
        }
        
        if (message.getTopic() == null) {
            message.setTopic(this.name);
        }
        
        messageQueue.offer(message);
        messageCount.incrementAndGet();
        logger.debug("Message {} sent to topic {}", message.getMessageId(), name);
        return message.getMessageId();
    }
    
    @Override
    public synchronized List<String> sendMessages(List<Message> messages) {
        checkClosed();
        
        if (messages == null || messages.isEmpty()) {
            throw new IllegalArgumentException("Messages cannot be null or empty");
        }
        
        List<String> messageIds = new ArrayList<>(messages.size());
        for (Message message : messages) {
            if (message.getTopic() == null) {
                message.setTopic(this.name);
            }
            messageQueue.offer(message);
            messageIds.add(message.getMessageId());
            messageCount.incrementAndGet();
        }
        
        logger.debug("Sent {} messages to topic {}", messages.size(), name);
        return messageIds;
    }
    
    @Override
    public synchronized Message pollMessage(String consumerId, long timeout) {
        checkClosed();
        
        if (!consumers.contains(consumerId)) {
            addConsumer(consumerId);
        }
        
        try {
            // 首先检查是否有未确认的消息可以重新发送
            Message unackMessage = getUnacknowledgedMessage(consumerId);
            if (unackMessage != null) {
                return unackMessage;
            }
            
            // 从队列中获取新消息
            Message message = messageQueue.poll(timeout, TimeUnit.MILLISECONDS);
            if (message != null) {
                // 检查消息是否已过期
                if (message.isExpired()) {
                    logger.debug("Message {} expired, skipping", message.getMessageId());
                    messageCount.decrementAndGet();
                    return pollMessage(consumerId, 0); // 递归获取下一条消息
                }
                
                // 将消息添加到未确认列表
                addToUnacknowledgedMessages(consumerId, message);
                logger.debug("Message {} polled by consumer {} from topic {}", 
                        message.getMessageId(), consumerId, name);
            }
            return message;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.warn("Polling message interrupted for consumer {} on topic {}", consumerId, name);
            return null;
        }
    }
    
    @Override
    public synchronized boolean acknowledge(String consumerId, String messageId) {
        checkClosed();
        
        Map<String, Message> consumerUnackMessages = unacknowledgedMessages.get(consumerId);
        if (consumerUnackMessages != null && consumerUnackMessages.remove(messageId) != null) {
            messageCount.decrementAndGet();
            logger.debug("Message {} acknowledged by consumer {} for topic {}", 
                    messageId, consumerId, name);
            
            // 如果消费者的未确认消息列表为空，则移除该列表
            if (consumerUnackMessages.isEmpty()) {
                unacknowledgedMessages.remove(consumerId);
            }
            return true;
        }
        
        logger.warn("Failed to acknowledge message {} for consumer {} on topic {}", 
                messageId, consumerId, name);
        return false;
    }
    
    @Override
    public int getMessageCount() {
        return messageCount.get();
    }
    
    @Override
    public int getConsumerCount() {
        return consumers.size();
    }
    
    @Override
    public boolean addConsumer(String consumerId) {
        checkClosed();
        
        if (consumerId == null || consumerId.isEmpty()) {
            throw new IllegalArgumentException("Consumer ID cannot be null or empty");
        }
        
        boolean added = consumers.add(consumerId);
        if (added) {
            logger.info("Consumer {} added to topic {}", consumerId, name);
        }
        return added;
    }
    
    @Override
    public boolean removeConsumer(String consumerId) {
        checkClosed();
        
        if (consumerId == null || consumerId.isEmpty()) {
            throw new IllegalArgumentException("Consumer ID cannot be null or empty");
        }
        
        boolean removed = consumers.remove(consumerId);
        if (removed) {
            // 移除消费者的未确认消息列表
            Map<String, Message> consumerUnackMessages = unacknowledgedMessages.remove(consumerId);
            if (consumerUnackMessages != null && !consumerUnackMessages.isEmpty()) {
                // 将未确认的消息放回队列
                for (Message message : consumerUnackMessages.values()) {
                    messageQueue.offer(message);
                }
                logger.info("Consumer {} removed from topic {}, {} unacknowledged messages requeued", 
                        consumerId, name, consumerUnackMessages.size());
            }
            logger.info("Consumer {} removed from topic {}", consumerId, name);
        }
        return removed;
    }
    
    @Override
    public synchronized void close() {
        if (closed) {
            return;
        }
        
        closed = true;
        consumers.clear();
        unacknowledgedMessages.clear();
        messageQueue.clear();
        messageCount.set(0);
        logger.info("Topic '{}' closed", name);
    }
    
    // 检查主题是否已关闭
    private void checkClosed() {
        if (closed) {
            throw new IllegalStateException("Topic '" + name + "' is closed");
        }
    }
    
    // 获取消费者的未确认消息
    private Message getUnacknowledgedMessage(String consumerId) {
        Map<String, Message> consumerUnackMessages = unacknowledgedMessages.get(consumerId);
        if (consumerUnackMessages != null && !consumerUnackMessages.isEmpty()) {
            // 返回第一条未确认的消息（按插入顺序）
            return consumerUnackMessages.values().iterator().next();
        }
        return null;
    }
    
    // 将消息添加到未确认列表
    private void addToUnacknowledgedMessages(String consumerId, Message message) {
        unacknowledgedMessages.computeIfAbsent(consumerId, k -> new LinkedHashMap<>())
                             .put(message.getMessageId(), message);
    }
}